// Package connection simulates part or similar interfaces  of TCP connection
package connection

import (
	"encoding/binary"
	"fmt"
	"go.uber.org/zap"
	gseTErr "gse/transport/errors"
	gsePacket "gse/transport/udp/packet"
	"hash/crc32"
	"net"
	"sync"
	"time"
)

const DefaultKeepAlive = 3000

// PACKET_SEND_RATE limits the count of calling Send api within a second
const DefaultPacketSendRate = 42000

// in milliseconds
const DefaultAckTimeout = 3000

const MaxDataLength = 1 << 32

// DefaultResendInterval in milliseconds.
const DefaultSendRetryInterval = 200
const DefaultSendRetries = 3

const SeqDedupDepth = 10

type Connection struct {
	remoteAddr        net.Addr
	lastHeartbeatTime time.Time
	// packetChan the incoming gse UDP data packet chan.
	packetChan chan gsePacket.UdpPacket
	// obPacketChan outband metadata chan.
	obPacketChan          chan *gsePacket.PacketHeader
	expirationTimer       *time.Timer
	closed                chan bool
	readTimer             *time.Timer
	readTimerRunning      bool
	logger                *zap.Logger
	OnClose               func(connection *Connection)
	udpConnection         *net.UDPConn
	srvMode               bool
	nextPacketSN          uint16
	nextPacketSNLock      *sync.Mutex
	sendLock              *sync.Mutex
	cfg                   *UDPConnectionConfig
	receiver              *UDPPacketReceiver
	deDupSeqList          [SeqDedupDepth]gsePacket.UdpPacket
	deDupSeqListNextIndex int
	deDupSeqListLock      *sync.Mutex
	// OnConnErr will be called when the connection is broken, e.g. the UDP server is down.
	OnConnErr func(error)
}

type UDPConnectionConfig struct {
	// PacketSendRate only available without ack
	PacketSendRate int
	// AckTimeout only available with ack. in milliseconds
	AckTimeout  int
	SendRetries int
	// in milliseconds.
	SendRetryInterval int
	// KeepAlive in millisecond
	KeepAlive int
	Logger    *zap.Logger
}

func NewDefaultUDPConnectionConfig() *UDPConnectionConfig {
	logger, _ := zap.NewDevelopment()
	return &UDPConnectionConfig{
		PacketSendRate:    DefaultPacketSendRate,
		AckTimeout:        DefaultAckTimeout,
		SendRetries:       DefaultSendRetries,
		SendRetryInterval: DefaultSendRetryInterval,
		KeepAlive:         DefaultKeepAlive,
		Logger:            logger,
	}
}

// Connect connects to a server.
func Connnect(remoteAddr net.Addr) (*Connection, error) {
	return newConnection(remoteAddr, nil)
}

// NewIncomingConnection wraps the remote(client) address and the UDP server's socket object into a TCP like Connection.
func NewIncomingConnection(remoteAddr net.Addr, udpSrvConn *net.UDPConn) (*Connection, error) {
	return newConnection(remoteAddr, udpSrvConn)
}

// newConnection when udpSrvConn is not nil, the Connection is created in the server side.
func newConnection(remoteAddr net.Addr, udpSrvConn *net.UDPConn) (*Connection, error) {
	c := &Connection{
		remoteAddr:        remoteAddr,
		lastHeartbeatTime: time.Now(),
		packetChan:        make(chan gsePacket.UdpPacket, 1000),
		obPacketChan:      make(chan *gsePacket.PacketHeader, 1000),
		expirationTimer:   time.NewTimer(time.Millisecond * time.Duration(DefaultKeepAlive)),
		closed:            make(chan bool),
		readTimer:         time.NewTimer(1),
		udpConnection:     udpSrvConn,
		sendLock:          new(sync.Mutex),
		cfg:               NewDefaultUDPConnectionConfig(),
		nextPacketSN:      1,
		deDupSeqListLock:  new(sync.Mutex),
		nextPacketSNLock:  new(sync.Mutex),
	}
	c.logger = c.cfg.Logger
	if udpSrvConn != nil {
		c.srvMode = true
	} else {
		var err error
		udpAdd, err := net.ResolveUDPAddr(remoteAddr.Network(), remoteAddr.String())
		if err != nil {
			return nil, err
		}
		udpSrvConn, err = net.DialUDP(udpAdd.Network(), nil, udpAdd)
		c.udpConnection = udpSrvConn
		if err != nil {
			return nil, err
		}
	}
	c.readTimer.Stop()

	if c.srvMode {
		go c.CheckExpiration()
	} else {
		c.receiver = &UDPPacketReceiver{
			closed:             c.closed,
			packetSize:         gsePacket.PacketSize,
			readDataBufferPool: make(chan []byte, 10),
			conn:               c.udpConnection,
			packetChan:         c.packetChan,
			obPacketChan:       c.obPacketChan,
			logger:             c.logger,
		}
		c.receiver.onConnErr = func(err error) {
			c.logger.Error(err.Error())
			c.onConnErr(err)
		}
		go c.receiver.RecvUDPPacket()
	}
	return c, nil
}

func (conn *Connection) KeepAlive() {
	go func() {
		d := gsePacket.NewHeartbeat()
		ttl := conn.cfg.KeepAlive - 100
		if ttl <= 0 {
			ttl = 100
		}
		ticker := time.NewTicker(time.Millisecond * time.Duration(ttl))
		for {
			select {
			case <-conn.closed:
				ticker.Stop()
				return
			case <-ticker.C:
				_, err := conn.internalSend(d)
				if err != nil {
					return
				}
			}
		}
	}()
}

func (conn *Connection) Ping() (responseDelay time.Duration, err error) {
	sn := conn.genPacketSN()
	d := gsePacket.NewPing(sn)
	st := time.Now()
	defer func() {
		responseDelay = time.Now().Sub(st)
	}()
	_, err = conn.internalSend(d)
	if err != nil {
		return
	}
	err = conn.waitAck(sn)
	return
}

func (conn *Connection) RemoteAddr() net.Addr {
	return conn.remoteAddr
}

func (conn *Connection) LocalAddr() net.Addr {
	return conn.udpConnection.LocalAddr()
}

func (conn *Connection) GetConfig() *UDPConnectionConfig {
	return conn.cfg
}
func (conn *Connection) SetConfig(cfg *UDPConnectionConfig) error {
	if cfg == nil {
		return nil
	}
	conn.sendLock.Lock()
	conn.cfg = cfg
	conn.sendLock.Unlock()
	if !conn.expirationTimer.Stop() {
		<-conn.expirationTimer.C
	}
	conn.expirationTimer.Reset(time.Millisecond * time.Duration(conn.cfg.KeepAlive))
	conn.logger = cfg.Logger
	return nil
}

func (conn *Connection) SetWriteBuffer(bytes int) error {
	return conn.udpConnection.SetWriteBuffer(bytes)
}

func (conn *Connection) SetReadBuffer(bytes int) error {
	return conn.udpConnection.SetReadBuffer(bytes)
}

func (conn *Connection) onConnErr(err error) {
	conn.resetDedupList()
	if conn.OnConnErr != nil {
		conn.OnConnErr(err)
	}
}
func (conn *Connection) resetDedupList() {
	conn.deDupSeqListLock.Lock()
	for i := range conn.deDupSeqList {
		conn.deDupSeqList[i] = nil
	}
	conn.deDupSeqListNextIndex = 0
	conn.deDupSeqListLock.Unlock()
}

func (conn *Connection) CheckExpiration() {
	idleTimeout := time.Millisecond * time.Duration(conn.cfg.KeepAlive)
	for {
		select {
		case <-conn.closed:
			return
		case t := <-conn.expirationTimer.C:
			select { // recheck the close state.
			case <-conn.closed:
				return
			default:
			}

			if t.After(conn.lastHeartbeatTime.Add(idleTimeout)) {
				conn.logger.Debug("connection expired, closing...",
					zap.String("remote_addr", conn.remoteAddr.String()), zap.Int("idle_time(ms)", conn.cfg.KeepAlive),
				)
				// send reset to remote peer
				conn.internalSend(gsePacket.NewReset())
				conn.Close()
				return
			} else {
				conn.expirationTimer.Reset(idleTimeout)
			}

		}
	}
}

func (conn *Connection) Close() {
	select {
	case <-conn.closed:
		return
	default:
	}
	conn.logger.Debug("closing connection...",
		zap.String("remote_addr", conn.remoteAddr.String()),
	)
	close(conn.closed)
	if conn.OnClose != nil {
		conn.OnClose(conn)
	}
	if !conn.srvMode {
		conn.udpConnection.Close()
	}
}

func (conn *Connection) updateHeartbeat() {
	conn.lastHeartbeatTime = time.Now()
}

// ReadData read a complete request/transaction data in block mode.
func (conn *Connection) ReadData() (data []byte, err error) {
	var payloadPackets [][]byte
	var payloadTotalLength uint32
	var payload []byte
	var dataHeadReceived bool
	var dataCRC uint32
	var payloadExpectedLength uint32
	st := time.Now()
READ:
	for {
		var p gsePacket.UdpPacket
		if conn.readTimer != nil && conn.readTimerRunning {
			select {
			case p = <-conn.packetChan:
				payload, err = p.GetPayload(), nil
				if err != nil {
					return nil, err
				}
			case <-conn.closed:
				return nil, gseTErr.ConectionClosed
			case <-conn.readTimer.C:
				return nil, gseTErr.ConnectionReadTimeout
			}
		} else {
			select {
			case p = <-conn.packetChan:
				payload, err = p.GetPayload(), nil
				if err != nil {
					return nil, err
				}
			case <-conn.closed:
				return nil, gseTErr.ConectionClosed
			}
		}

		if p.GetType() == gsePacket.PacketTypeReset {
			return nil, gseTErr.ConectionReset
		}

		conn.deDupSeqListLock.Lock()
		for i := 0; i < SeqDedupDepth; i++ {
			if conn.deDupSeqList[i] == nil {
				break
			}
			if conn.deDupSeqList[i].GetSeq() == p.GetSeq() {
				conn.logger.Debug("packet may be duplicated, dropped. ", zap.Uint16("seq", p.GetSeq()), zap.String("remote_addr", conn.remoteAddr.String()))
				conn.deDupSeqListLock.Unlock()
				continue READ
			}
		}
		conn.deDupSeqList[conn.deDupSeqListNextIndex] = p
		conn.deDupSeqListNextIndex += 1
		if conn.deDupSeqListNextIndex >= SeqDedupDepth {
			conn.deDupSeqListNextIndex = 0
		}
		conn.deDupSeqListLock.Unlock()

		if p.IsHead() {
			if dataHeadReceived {
				return nil, fmt.Errorf("data broken: data head already received! packet seq[%v], packet type[%v]", p.GetSeq(), p.GetType())
			}
			headPacket := p.(*gsePacket.HeadPacket)
			dataCRC = headPacket.CRC
			payloadExpectedLength = headPacket.DataLength
			dataHeadReceived = true
		} else if p.IsTail() {
			if !dataHeadReceived && p.GetType() != gsePacket.PacketTypeHeadWithTail {
				conn.logger.Warn(
					fmt.Errorf("head data is expected firstly, but tail received. data dropped! packet seq[%v], packet type[%v]", p.GetSeq(), p.GetType()).Error(),
					zap.String("remote_addr", conn.remoteAddr.String()),
				)
				continue
			}
		} else {
			if !dataHeadReceived {
				conn.logger.Warn(
					fmt.Errorf("head data is expected firstly, data dropped! packet seq[%v], packet type[%v]", p.GetSeq(), p.GetType()).Error(),
					zap.String("remote_addr", conn.remoteAddr.String()),
				)
				continue
			}
		}
		if len(payload) > 0 {
			payloadPackets = append(payloadPackets, payload)
			payloadTotalLength += uint32(len(payload))
			conn.logger.Debug("progress:", zap.Uint32("total", payloadExpectedLength),
				zap.Uint32("received", payloadTotalLength),
				zap.Uint32("remaining", payloadExpectedLength-payloadTotalLength),
				zap.Float32("percent", float32(payloadTotalLength)/float32(payloadExpectedLength)),
			)
		}
		if payloadTotalLength > payloadExpectedLength {
			return nil, fmt.Errorf("data broken: read data length incrrect! expected length[%v], crrently read bytes[%v]", payloadExpectedLength, payloadTotalLength)
		}

		if p.IsTail() {
			break
		}
	}
	data = make([]byte, payloadTotalLength)
	offset := 0
	for i := range payloadPackets {
		copy(data[offset:], payloadPackets[i])
		offset += len(payloadPackets[i])
	}
	checkedCRC := crc32.ChecksumIEEE(data)
	if checkedCRC != dataCRC {
		conn.logger.Warn("read bad data.",
			zap.Uint32("expected_length", payloadExpectedLength),
			zap.Uint32("expected_crc", dataCRC),
			zap.Uint32("actual_length", payloadTotalLength),
			zap.Uint32("actual_crc", checkedCRC),
		)
		return nil, gseTErr.BadDataChecksum
	}
	et := time.Now()
	conn.logger.Debug("read data done:",
		zap.Uint32("data_crc", checkedCRC),
		zap.String("remote_addr", conn.remoteAddr.String()),
		zap.String("time_cost", fmt.Sprintf("%v", et.Sub(st))),
		zap.String("transfer_rate", fmt.Sprintf("%v bytes/s", int(float64(payloadTotalLength)/et.Sub(st).Seconds()))),
	)
	return
}

// ReadData read a complete request/transaction blockly.
func (conn *Connection) SetReadDeadline(t time.Time) {
	conn.readTimer.Stop()
	zTime := time.Time{}
	if t == zTime {
		conn.readTimerRunning = false
		conn.readTimer.Stop()
		return
	}
	conn.readTimerRunning = true
	conn.readTimer.Reset(t.Sub(time.Now()))
}

func (conn *Connection) internalSend(data []byte) (n int, err error) {
	if conn.srvMode {
		return conn.udpConnection.WriteTo(data, conn.remoteAddr)
	} else {
		return conn.udpConnection.Write(data)
	}
}

// AllocateUDPPacket allocates the incoming packet to the corresponding connection handling routine.
func (conn *Connection) AllocateUDPPacket(udpPacket gsePacket.UdpPacket) error {
	if !conn.srvMode {
		fmt.Errorf(`"AllocateUDPPacket" only applys to server side connection`)
	}
	conn.updateHeartbeat()
	if udpPacket.GetType() == gsePacket.PacketTypeAck || udpPacket.GetType() == gsePacket.PacketTypeReset {
		select {
		case conn.obPacketChan <- udpPacket.(*gsePacket.PacketHeader):
			return nil
		case <-conn.closed:
			return gseTErr.ConectionClosed
		default:
			return fmt.Errorf("conn data ack recv chan is full, new packet will be dropped! chan cap[%v].", cap(conn.obPacketChan))
		}
	} else if udpPacket.GetType() != gsePacket.PacketTypeHeartbeat && udpPacket.GetType() != gsePacket.PacketTypePing {
		select {
		case conn.packetChan <- udpPacket:
		case <-conn.closed:
			return gseTErr.ConectionClosed
		default:
			return fmt.Errorf("conn data recv chan is full, new packet will be dropped! chan cap[%v].", cap(conn.packetChan))
		}
	}
	return nil
}

func (conn *Connection) genPacketSN() uint16 {
	conn.nextPacketSNLock.Lock()
	n := conn.nextPacketSN
	conn.nextPacketSN += 1
	if conn.nextPacketSN > gsePacket.MaxPacketSeqNo {
		conn.nextPacketSN = 1
	}
	conn.nextPacketSNLock.Unlock()
	return n
}

func (conn *Connection) SendData(data []byte) (int, error) {
	return conn.send(data, false)
}

func (conn *Connection) SendDataWithAck(data []byte) (int, error) {
	return conn.send(data, true)
}

func (conn *Connection) send(data []byte, ack bool) (int, error) {
	select {
	case <-conn.closed:
		return 0, gseTErr.ConectionClosed
	default:
	}
	dataLen := len(data)
	if dataLen > MaxDataLength {
		return 0, fmt.Errorf("data length execeeded max %v.", MaxDataLength)
	}
	conn.sendLock.Lock()
	defer conn.sendLock.Unlock()
	packet := make([]byte, gsePacket.PacketSize)
	packetLen := int(gsePacket.PacketSize)
	var PacketType gsePacket.PacketType

	dataCRC := crc32.ChecksumIEEE(data)
	var dataBytes []byte
	var header = make([]byte, 2)
	var totalSentN, totalDataSentN int
	for n := 0; n < dataLen; {
		headerInt := uint16(gsePacket.CURRENT_PROTO_ID) << 14
		if ack {
			headerInt += uint16(1) << 13
		}
		// the first packet
		if n == 0 {
			// 10: (fixed header(2) + variable header(8) length
			// the first packet size must be PACKET_SIZE
			if packetLen-10 >= dataLen {
				PacketType = gsePacket.PacketTypeHeadWithTail
				dataBytes = data
				n = dataLen
			} else {
				PacketType = gsePacket.PacketTypeHead
				n = packetLen - 10
				dataBytes = data[0:n]
			}
		} else {
			// 2: the fixed header length
			// the last packet of the data.
			if packetLen-2 >= dataLen-n {
				PacketType = gsePacket.PacketTypeTail
				dataBytes = data[n:]
				n = dataLen
			} else {
				PacketType = gsePacket.PacketTypeMid
				offset := n + packetLen - 2
				dataBytes = data[n:offset]
				n = offset
			}
		}
		headerInt += uint16(PacketType) << 10
		sn := conn.genPacketSN()
		headerInt += sn
		binary.BigEndian.PutUint16(header, headerInt)
		offsetT := 0
		copy(packet[0:], header)
		offsetT += 2
		// The first  packet
		if PacketType == 0 || PacketType == 3 {
			// put data length
			dataLengthBytes := make([]byte, 4)
			binary.BigEndian.PutUint32(dataLengthBytes, uint32(dataLen))
			copy(packet[offsetT:], dataLengthBytes)
			offsetT += 4

			// put crc32
			crcBytes := make([]byte, 4)
			binary.BigEndian.PutUint32(crcBytes, dataCRC)
			copy(packet[offsetT:], crcBytes)
			offsetT += 4
		}
		copy(packet[offsetT:], dataBytes)
		offsetT += len(dataBytes)
		conn.logger.Debug("sending packet...",
			zap.Uint16("seq", sn),
			zap.Int("total_sent", totalSentN),
			zap.Uint8("packet_type", uint8(PacketType)),
			zap.String("remoate_addr", conn.remoteAddr.String()),
		)

		var tn int
		var err error
		for n := conn.cfg.SendRetries; n >= 0; n-- {
			if n != conn.cfg.SendRetries { // already run into retry
				conn.logger.Debug("retry sending packet...", zap.Int("retries", conn.cfg.SendRetries-n),
					zap.Uint16("seq", sn),
					zap.String("remoate_addr", conn.remoteAddr.String()))
				time.Sleep(time.Millisecond * time.Duration(conn.cfg.SendRetryInterval))
			}
			tn, err = conn.internalSend(packet[0:offsetT])
			if err != nil {
				continue
			}

			if ack {
				err = conn.waitAck(sn)
				if err != nil {
					if err == gseTErr.ConectionReset || err == gseTErr.ConectionClosed {
						return totalSentN, err
					} else {
						err = fmt.Errorf("wait for ack[%v]  err: %v", sn, err)
					}
					continue
				}
			} else if conn.cfg.PacketSendRate > 0 {
				time.Sleep(time.Second / time.Duration(conn.cfg.PacketSendRate))
			}
			break
		}

		totalSentN += tn
		if err != nil {
			return totalSentN, err
		} else if tn != offsetT {
			return totalSentN, fmt.Errorf("packet sent incomplete! expects %v, but sent %v", offsetT, tn)
		}
		totalDataSentN += len(dataBytes)
	}
	return totalDataSentN, nil
}

func (conn *Connection) waitAck(sn uint16) error {
	conn.logger.Debug("waiting ack...", zap.Uint16("seq", sn))
	timeout := time.After(time.Millisecond * time.Duration(conn.cfg.AckTimeout))
ReadAck:
	select {
	case <-conn.closed:
		return gseTErr.ConectionClosed
	case <-timeout:
		return fmt.Errorf("wait ack for seq[%v] timedout: %v", sn, conn.cfg.AckTimeout)
	case obPacket := <-conn.obPacketChan:
		if obPacket.GetType() == gsePacket.PacketTypeReset {
			return gseTErr.ConectionReset
		}
		if obPacket.Seq != sn {
			conn.obPacketChan <- obPacket
			goto ReadAck
		}
	}
	conn.logger.Debug("ack received.", zap.Uint16("seq", sn))
	return nil
}
